package com.bw.yk02.fnc;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;

public class MySQLSourceTest implements DebeziumDeserializationSchema<String> {


    @Override
    public void deserialize(SourceRecord c, Collector collector) throws Exception {

        JSONObject jsonObject = new JSONObject();
        String topic=c.topic();
        String [] strs=topic.split("\\.");
        String db=strs[1];
        String table=strs[2];

        Struct struct = (Struct) c.value();


        if(struct!=null)
        {
            String op = struct.getString("op");
            Struct before = struct.getStruct("before");
            Struct after = struct.getStruct("after");

            jsonObject.put("db", db);
            jsonObject.put("table", table);
            jsonObject.put("op", op);

            JSONObject date = new JSONObject();

            if(after!=null)
            {
                for (Field field : after.schema().fields()) {
                    String fieldName = field.name();
                    Object value = after.get(fieldName);
                    value = checkType(fieldName, value);
                    date.put(fieldName, value);
                }
            }

            JSONObject old = new JSONObject();

            if(before!=null)
            {
                for (Field field : before.schema().fields()) {
                    String fieldName = field.name();
                    Object value = before.get(fieldName);
                    value = checkType(fieldName, value);
                    old.put(fieldName, value);
                }
            }

            jsonObject.put("data", date);
            jsonObject.put("old", old);
        }

        collector.collect(jsonObject.toJSONString());
    }

    @Override
    public TypeInformation getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

    public static Object checkType(String c,Object v) {

        if(c.endsWith("_time") && v!=null && v instanceof Long){
            Timestamp dateTimeValue =new Timestamp((Long) v);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            v = sdf.format(dateTimeValue);
        }
        return v;
    }
}
